package com.xiongjie.SingleLimit.algorithm;

import com.google.common.util.concurrent.RateLimiter;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.IntStream;

/**
 * 使用漏桶算法实现限流功能。关键是每次请求过来时，判断这段时间流出的水量，当这个新的请求进去时，是否在漏桶中
 * 如果在，则请求继续，否则请求超出漏桶容量，请求丢弃
 * <p>
 * 降级策略就是抛异常，存入消息队列或者存入数据库，延迟处理
 * <p>
 * 使用RateLimiter，有种消息队列的格式
 */
public class LBLimiterWithGuava implements Runnable {

    //使用并行链表队列定义漏桶
    private final ConcurrentLinkedQueue<Integer> container = new ConcurrentLinkedQueue<>();
    private final RateLimiter rateLimiter = RateLimiter.create(10, 1, TimeUnit.SECONDS);
    final AtomicInteger MONIDATA = new AtomicInteger(0);

    private ReentrantLock lock = new ReentrantLock(true);   //并发锁

    /**
     * 1.往桶里面写数据，来一个请求就判断，存储请求
     *
     * @param data
     */
    public void submit(Integer data) {
        final ReentrantLock lock = this.lock;
        lock.lock();

        try{
            if (container.size() < 50) {
                container.offer(data);
                System.out.println("提交数据：" + data + "队列大小=" + container.size());
            } else {
                //这里时候采用降级策略了。消费速度跟不上产生速度时，而且桶满了，抛出异常，或者存入MQ DB等后续处理
                new IllegalStateException("漏桶已满，请稍后再试");
            }
        }finally {
            lock.unlock();
        }
    }

    @Override
    public void run() {
        if (!container.isEmpty()) {
            rateLimiter.acquire();
            Integer data = container.poll();    //从队列取出请求处理

            //继续执行请求之后的操作
            System.out.println("开始消费：" + data);

        } else {
            //当木桶的消费完后，可以消费那些降级存入MQ或者DB里面的数据
            System.out.println("将会消费降级后的数据，从消息队列或者数据库中");
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {

        final LBLimiterWithGuava bucket = new LBLimiterWithGuava();
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutorService.scheduleAtFixedRate(bucket, 100, 100, TimeUnit.MILLISECONDS);

        //模拟启动两个线程不停的发送请求
        IntStream.range(0, 2).forEach(i -> {
            new Thread(() -> {
                while (true) {
                    int data = bucket.MONIDATA.incrementAndGet();
                    try {
                        bucket.submit(data);
                        TimeUnit.MILLISECONDS.sleep(300);
                    } catch (Exception e) {
                        //对submit时，如果桶满了可能会抛出异常
                        if (e instanceof IllegalStateException) {
                            System.out.println(e.getMessage());
                            //当满了后，生产线程就休眠1分钟
                            try {
                                System.out.println("漏桶已满，线程睡眠一分钟");
                                TimeUnit.SECONDS.sleep(60);
                            } catch (InterruptedException e1) {
                                e1.printStackTrace();
                            }
                        }
                    }
                }
            }).start();
        });
    }

}
